package com.shujia.test

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.{EnvironmentSettings, _}
import org.apache.flink.table.api.bridge.scala.{StreamTableEnvironment, _}
import org.apache.flink.table.catalog.hive.HiveCatalog
import org.apache.flink.types.Row

object testDemo1 {
  def main(args: Array[String]): Unit = {
    val bsEnv = StreamExecutionEnvironment.getExecutionEnvironment
    val bsSetting = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()

    val bsTable = StreamTableEnvironment.create(bsEnv, bsSetting)
    val hiveCatalog = new HiveCatalog("myHive", "sent", "src/main/resources")

    bsTable.registerCatalog("myHive", hiveCatalog)
    bsTable.useCatalog("myHive")

    val table: Table = bsTable.sqlQuery(
      """
        |select * from ods.ods_kafka_comment limit 10
        |""".stripMargin)

    table.toAppendStream[Row].print()

    bsEnv.execute()

  }

}
